use crate::cache::IndexChange;
use crate::keys::VddKey;
use crate::record_db::UniKeyTraitExtern;
use crate::value_db::UniqueChange;
use crate::{
    cache, count_in_key_head, delete_first_child, load_by_key_head_with_key, log_db, record_db,
    record_rule_db, subscribe, value_db, ROCKSDB,
};
use rust_rocksdb::WriteBatchWithTransaction;
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use visit_dd_core::i18n_text::E_ROCKSDB;
use visit_dd_core::id_factory::ID_FACTORY;
use visit_dd_core::proj::VddLogPart;
use visit_dd_core::record::{VddRecord, VersionedRecord};
use visit_dd_core::value::VddValueRef;
use visit_dd_core::{
    alog, extract_last_str_from_key, i18n, Ares, Astr, FieldId, RecordId, AC_EDIT,
};

/// 将现有的记录值保存为一个版本，并保持版本数量小于100
pub(crate) async fn create_version(
    batch: &mut WriteBatchWithTransaction<true>,
    record: &VddRecord,
) -> u64 {
    let version = ID_FACTORY.lock().unwrap().lazy_generate();
    // 将 field value map 保存为一个版本
    let vk = VddKey::BoundRecordAndVersion(record.meta.id.clone(), version);
    batch.put_cf(
        vk.db_name().unwrap(),
        vk.to_string(),
        json!(&record.value).to_string(),
    );
    // 版本数限制为100
    let key_head = VddKey::BoundRecordAndVersion(record.meta.id.clone(), 0);
    if count_in_key_head(&key_head, |_| true) >= 100 {
        delete_first_child(batch, &key_head);
    }
    version
}

pub fn count_version(record_id: &RecordId) -> u8 {
    let key = VddKey::BoundRecordAndVersion(record_id.clone(), 0);
    count_in_key_head(&key, |_| true) as u8
}

pub fn select_versions(record_id: &RecordId) -> Vec<Arc<VersionedRecord>> {
    let mut r = Vec::<Arc<VersionedRecord>>::new();
    let key_head = VddKey::BoundRecordAndVersion(record_id.clone(), 0);
    load_by_key_head_with_key(&key_head).iter().for_each(|x| {
        // rocksdb 中的 value 是 fieldvalue 的 hashmap
        let record_value = serde_json::from_slice::<HashMap<FieldId, VddValueRef>>(&x.1).unwrap();
        let version = extract_last_str_from_key(&x.0).parse::<u64>().unwrap();
        r.push(Arc::new(VersionedRecord {
            version,
            record_id: record_id.to_owned(),
            value: record_value,
        }));
    });
    r
}

pub fn delete_version(record_id: &RecordId, version: u64) -> Ares {
    let key = VddKey::BoundRecordAndVersion(record_id.clone(), version).to_string();
    ROCKSDB.delete(key).map_err(|e| e.to_string().into())
}

/// 加载某个版本
fn load_by_version(
    record_id: &RecordId,
    version: u64,
) -> Result<HashMap<FieldId, VddValueRef>, Astr> {
    let vk = VddKey::BoundRecordAndVersion(record_id.clone(), version);
    if let Ok(Some(vu8)) = ROCKSDB.get_cf(vk.db_name().unwrap(), vk.to_string()) {
        Ok(serde_json::from_slice::<HashMap<FieldId, VddValueRef>>(&vu8).unwrap())
    } else {
        Err(i18n!(E_ROCKSDB))
    }
}

/// 回滚版本
pub async fn roll_back(model_id: u64, record_id: &RecordId, version: u64) -> Ares {
    let schema = cache::get_schema(model_id).await?;
    let rec = record_db::load_record(&schema, record_id, &[]).unwrap();
    let v_rules = record_rule_db::rr_select_in_model(rec.meta.model_id).await;
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    rec.delete_unique_after_delete_record(&v_rules, &mut unique_changes);
    let meta = record_db::load_meta_from_rdb(record_id).expect(i18n!(E_ROCKSDB).as_ref());
    let fvs = load_by_version(record_id, version)?;
    let mut rec = VddRecord {
        meta: Arc::new(meta),
        value: fvs,
    };
    record_db::save(&mut rec, &schema, &v_rules, false).await?;
    subscribe::send_change_event(
        &mut batch,
        &mut index_changes,
        rec.meta.model_id,
        &rec.meta.id,
        AC_EDIT,
    );
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(rec.meta.uid.clone(), "回滚记录".into()),
            VddLogPart::LpModel(rec.meta.model_id),
            VddLogPart::LpRecord(rec.meta.id.to_owned()),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        value_db::apply_unique_change(&unique_changes)?;
    }
    ares
}
